
package hasthi.manager.monitering;

import hasthi.common.HasthiException;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.HasthiConstants;
import hasthi.manager.ManagementLogMessageDocument;
import hasthi.manager.ManagementLogMessageDocument.ManagementLogMessage;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.StringReader;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xmlbeans.XmlObject;

import wsmg.NotificationHandler;
import wsmg.WseClientAPI;

/**
 * Start listening to a given topic, and call a trigger action when a matching notification arrived
 * @author Srinath Perera(hperera@cs.indiana.edu)
 */

public class MoniteringAgent implements NotificationHandler{
    private Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);
    private String brokerUrl;
    private String topic;
    private final WseClientAPI client;
    private List<String> subcriptionIds = new ArrayList<String>();
    private List<EventListener> triggers =  new ArrayList<EventListener>();
    
    public MoniteringAgent(String brokerUrl,String topic){
        this.brokerUrl = brokerUrl;
        this.topic = topic;
        client = new WseClientAPI();
    }
    
    public void startMonitering()throws HasthiException{
        final NotificationHandler notificationHandler = this;
        Runnable runnable = new Runnable() {
            public void run() {
                try {
                    //int port = 34677;
                    int port  = client.startConsumerService(0, notificationHandler);
                    
                    String consumerUrl = "http://"+InetAddress.getLocalHost().getCanonicalHostName()+":"+port;
                    int count = 0;
                    Exception finalError = null;
                    while(count < 5){
                        try{
                            String subscribtionID = client.subscribe(brokerUrl, consumerUrl, topic);
                            subcriptionIds.add(subscribtionID);
                            System.out.println("Listen to events id= "+  subscribtionID + "##############");
                            cleanupOldSubscriptions(subscribtionID);
                            return;
                        }catch(Exception e){
                            finalError = e;
                            e.printStackTrace();
                            try {
                                Thread.sleep(60*1000);
                            } catch (InterruptedException e1) {e1.printStackTrace();}
                            count++;
                        }
                    }
                    if(finalError != null){
                        log.error(finalError);
                    }
                } catch (UnknownHostException e) {
                    log.error(e);
                }
            }
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }
    
    public void stopMonitoring()throws HasthiException{
        for(String subcriptionId:subcriptionIds){
            client.unSubscribe(brokerUrl, subcriptionId);
        }
        client.shutdownConsumerService();
    }
    
    public void addTrigger(EventListener trigger){
        triggers.add(trigger);
    }
    
    private void cleanupOldSubscriptions(String subscriptionid){
        try {
            String userHome = System.getProperty("user.home");
            File subscriptionsFile = new File(userHome,".hasthi_config/subscriptions");
            if(subscriptionsFile.exists()){
                String subscriptionsAsStr = Utils.readFile(subscriptionsFile.getAbsolutePath());
                String[] subscriptions = subscriptionsAsStr.split("@@@");
                for(String subcriptionId:subscriptions){
                    try {
                        if(subcriptionId.trim().length() > 0){
                            client.unSubscribe(brokerUrl, subcriptionId);    
                            log.info("clean old subscription "+subcriptionId);
                        }
                    } catch (Throwable e) {/*log.error(e);*/}
                }
            }else{
                subscriptionsFile.getParentFile().mkdir();
            }
            try {
            FileWriter out = new FileWriter(subscriptionsFile);
            out.write("@@@");
            out.write(subscriptionid);
            out.close();
            } catch (Throwable e) {
                log.error(e);
            }
        } catch (IOException e) {
            log.error(e);
        }
    } 
    
    public void handleNotification(String messageAsStr) {
        try {
            XmlObject message = XmlObject.Factory.parse(new StringReader(messageAsStr));
            XmlObject bodyContent = XmlBeanUtils.extractBodyContent(message);
            //This is to make sure, Xml Beans document is create from XML fragment 
            //bodyContent = XmlObject.Factory.parse(bodyContent.newXMLStreamReader());
            for(EventListener trigger:triggers){
                try{
                    if(trigger.getSearchPath() != null){
                        XmlObject[] matchingResults = bodyContent.selectPath(trigger.getSearchPath());
                        if(matchingResults != null && matchingResults.length > 0){
                            trigger.performTriggerAction(bodyContent,matchingResults);
                        }
                    }else{
                        trigger.performTriggerAction(bodyContent,new XmlObject[]{bodyContent});
                    }
                }catch(Exception e){
                    log.error(e);
                }
            }
            //System.out.println(bodyContent);
        } catch (Exception e) {
            log.error(e);
        }
    }
    public static void main(String[] args) throws Exception {
        final FileWriter out = new FileWriter("leadevents.log",true);
        
        MoniteringAgent moniteringAgent = new MoniteringAgent("http://tyr14.cs.indiana.edu:12346/",">");
        moniteringAgent.addTrigger(new EventListener() {
            public void performTriggerAction(XmlObject event, XmlObject[] matchingResults)
                    throws HasthiException {
                try {
                    out.write("#####\n");
                    String eventAsXml = event.xmlText();
                    System.out.println(eventAsXml);
                    out.write(eventAsXml);
                    out.write("\n");
                    out.flush();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            public String getSearchPath() {
                return null;
            }
        });
        moniteringAgent.startMonitering();
    }
}

